package cn.smileyan.demos;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.MultipleParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.apache.flink.util.Preconditions;


/**
 * 基于 DataStream API 的单词统计
 * 通过命令行参数指定输入文件路径（--input /usr/local/hello.txt），打印单词的个数
 * @author Smileyan
 */
@Slf4j
public class StreamWordCount {
    /**
     * 默认的用于统计单词个数的字符串
     */
    public static final String DEFAULT_WORDS = "Flink’s Table & SQL API makes it possible to work with queries written " +
            "in the SQL language, but these queries need to be embedded within a table program that is written in either Java or Scala. " +
            "Moreover, these programs need to be packaged with a build tool before being submitted to a cluster. " +
            "This more or less limits the usage of Flink to Java/Scala programmers" +
            "The SQL Client aims to provide an easy way of writing, debugging, and submitting table programs " +
            "to a Flink cluster without a single line of Java or Scala code. " +
            "The SQL Client CLI allows for retrieving and visualizing real-time results from the running distributed " +
            "application on the command line.";

    public static void main(String[] args) throws Exception {
        final MultipleParameterTool params = MultipleParameterTool.fromArgs(args);
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度
        env.getConfig().setParallelism(3);
        // 将全局参数传递给执行环境
        env.getConfig().setGlobalJobParameters(params);

        DataStream<String> text = null;
        // 根据输入参数判断是否指定了输入文件路径
        if (params.has("input")) {
            // 遍历所有输入文件路径，将它们的数据合并为一个数据流
            for (String input : params.getMultiParameterRequired("input")) {
                if (text == null) {
                    text = env.readTextFile(input);
                } else {
                    text = text.union(env.readTextFile(input));
                }
            }
            Preconditions.checkNotNull(text, "Input DataStream should not be null.");
        } else {
            // 使用默认单词集合
            log.info("use default words");
            text = env.fromElements(DEFAULT_WORDS);
        }

        // 对文本数据进行分词并计数
        assert text != null;
        DataStream<Tuple2<String, Integer>> counts = text.flatMap(new Tokenizer())
                .keyBy(value -> value.f0)
                .sum(1);

        // 打印结果到标准输出
        log.info("Printing result to stdout. Use --output to specify output path.");
        counts.print();

        // 执行作业
        env.execute("Streaming WordCount");
    }

    /**
     * 分词函数，实现了 FlatMapFunction 接口。
     * 将输入的文本行分割为单词，并为每个单词生成一个键值对（单词，1）。
     */
    public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {

        private static final long serialVersionUID = 8061659867139246041L;

        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            // 将文本行转换为小写并按非单词字符分割
            String[] tokens = value.toLowerCase().split("\\W+");

            // 遍历分割后的单词数组，将每个单词生成键值对并输出到结果收集器
            for (String token : tokens) {
                if (!token.isEmpty()) {
                    out.collect(Tuple2.of(token, 1));
                }
            }
        }
    }
}
